package com.zenithmind.common.utils;

import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

/**
 * RocketMQ工具类
 * 封装消息发送的常用方法
 */
@Component
public class MQUtils {
    private static final Logger logger = LoggerFactory.getLogger(MQUtils.class);

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 同步发送消息
     *
     * @param topic   主题
     * @param message 消息体
     * @return 发送结果
     */
    public SendResult syncSend(String topic, Object message) {
        try {
            return rocketMQTemplate.syncSend(topic, message);
        } catch (Exception e) {
            logger.error("同步发送消息失败: topic={}, message={}, error={}", topic, message, e.getMessage(), e);
            throw e;
        }
    }

    /**
     * 同步发送消息（带标签）
     *
     * @param topic   主题
     * @param tag     标签
     * @param message 消息体
     * @return 发送结果
     */
    public SendResult syncSend(String topic, String tag, Object message) {
        try {
            return rocketMQTemplate.syncSend(topic + ":" + tag, message);
        } catch (Exception e) {
            logger.error("同步发送消息失败: topic={}, tag={}, message={}, error={}", topic, tag, message, e.getMessage(), e);
            throw e;
        }
    }

    /**
     * 异步发送消息
     *
     * @param topic      主题
     * @param message    消息体
     * @param callback   回调函数
     */
    public void asyncSend(String topic, Object message, SendCallback callback) {
        try {
            rocketMQTemplate.asyncSend(topic, message, callback);
        } catch (Exception e) {
            logger.error("异步发送消息失败: topic={}, message={}, error={}", topic, message, e.getMessage(), e);
            throw e;
        }
    }

    /**
     * 异步发送消息（带标签）
     *
     * @param topic      主题
     * @param tag        标签
     * @param message    消息体
     * @param callback   回调函数
     */
    public void asyncSend(String topic, String tag, Object message, SendCallback callback) {
        try {
            rocketMQTemplate.asyncSend(topic + ":" + tag, message, callback);
        } catch (Exception e) {
            logger.error("异步发送消息失败: topic={}, tag={}, message={}, error={}", topic, tag, message, e.getMessage(), e);
            throw e;
        }
    }

    /**
     * 单向发送消息（不关心发送结果）
     *
     * @param topic   主题
     * @param message 消息体
     */
    public void sendOneWay(String topic, Object message) {
        try {
            rocketMQTemplate.sendOneWay(topic, message);
        } catch (Exception e) {
            logger.error("单向发送消息失败: topic={}, message={}, error={}", topic, message, e.getMessage(), e);
            throw e;
        }
    }

    /**
     * 单向发送消息（带标签，不关心发送结果）
     *
     * @param topic   主题
     * @param tag     标签
     * @param message 消息体
     */
    public void sendOneWay(String topic, String tag, Object message) {
        try {
            rocketMQTemplate.sendOneWay(topic + ":" + tag, message);
        } catch (Exception e) {
            logger.error("单向发送消息失败: topic={}, tag={}, message={}, error={}", topic, tag, message, e.getMessage(), e);
            throw e;
        }
    }

    /**
     * 发送延迟消息
     *
     * @param topic      主题
     * @param message    消息体
     * @param delayLevel 延迟级别
     * @return 发送结果
     */
    public SendResult syncSendDelayMessage(String topic, Object message, int delayLevel) {
        try {
            Message<Object> msg = MessageBuilder.withPayload(message).build();
            return rocketMQTemplate.syncSend(topic, msg, 3000, delayLevel);
        } catch (Exception e) {
            logger.error("发送延迟消息失败: topic={}, message={}, delayLevel={}, error={}", topic, message, delayLevel, e.getMessage(), e);
            throw e;
        }
    }

    /**
     * 发送延迟消息（带标签）
     *
     * @param topic      主题
     * @param tag        标签
     * @param message    消息体
     * @param delayLevel 延迟级别
     * @return 发送结果
     */
    public SendResult syncSendDelayMessage(String topic, String tag, Object message, int delayLevel) {
        try {
            Message<Object> msg = MessageBuilder.withPayload(message).build();
            return rocketMQTemplate.syncSend(topic + ":" + tag, msg, 3000, delayLevel);
        } catch (Exception e) {
            logger.error("发送延迟消息失败: topic={}, tag={}, message={}, delayLevel={}, error={}", topic, tag, message, delayLevel, e.getMessage(), e);
            throw e;
        }
    }
} 